package cn.emailChat.ai.service.scheduler;

import cn.emailChat.ai.common.enums.MessageStatusType;
import cn.emailChat.ai.domain.Email;
import cn.emailChat.ai.domain.MessageStatus;
import cn.emailChat.ai.dto.MailInboundEvent;
import cn.emailChat.ai.repo.EmailRepo;
import cn.emailChat.ai.repo.MessageStatusRepo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Date;
import java.util.List;
import java.util.Optional;

/**
 * @Description 扫描消息状态表，重试失败的消息，删除完成的消息（分区投递版）
 * @Author susu
 * @Date 2025/8/18
 */
@Slf4j
@Service
public class MessageRetry {

    private final RabbitTemplate rabbitTemplate;
    private final MessageStatusRepo messageStatusRepo;
    private final EmailRepo emailRepo;

    /* ====== 分区投递配置（方案B）====== */
    @Value("${mail.reply.exchange:auto.reply.x}")
    private String replyExchange;                         // 交换机名称

    @Value("${mail.reply.routing-prefix:auto.reply.p}")
    private String replyRoutingPrefix;                    // routingKey 前缀：auto.reply.p0/1/2

    @Value("${mail.reply.partition.count:3}")
    private int partitionCount;                           // 分区数量：3

    // 用于构造 MailInboundEvent 的收件人（你的接收邮箱/账户）
    @Value("${mail.imap.username}")
    private String toEmail;

    /* ========== 重试配置 ========== */
    @Value("${mail.status.retry.enabled:true}")
    private boolean retryEnabled;

    @Value("${mail.status.retry.max-retries:3}")
    private int maxRetries;

    @Value("${mail.status.retry.batch-size:100}")
    private int batchSize;

    public MessageRetry(RabbitTemplate rabbitTemplate, MessageStatusRepo messageStatusRepo, EmailRepo emailRepo) {
        this.rabbitTemplate = rabbitTemplate;
        this.messageStatusRepo = messageStatusRepo;
        this.emailRepo = emailRepo;
    }

    // 间隔秒：默认每 120 秒拉一批失败记录
    @Scheduled(fixedDelayString = "#{${mail.status.retry.interval-seconds:120} * 1000}")
    @Transactional
    public void retryFailedMessages() {
        if (!retryEnabled) return;

        Pageable page = PageRequest.of(0, Math.max(1, batchSize));
        List<MessageStatus> failedList = messageStatusRepo
                .findByStatusAndRetryCountLessThanOrderByUpdateTimeAsc(
                        MessageStatusType.FAILED, maxRetries, page);

        if (failedList.isEmpty()) return;
        log.info("[Retry] 准备重试失败消息 {} 条", failedList.size());

        for (MessageStatus ms : failedList) {
            String messageId = ms.getMessageId();
            LocalDateTime now = LocalDateTime.now();

            // 载入关联 Email
            Optional<Email> emailOpt = emailRepo.findByMessageId(messageId);
            if (emailOpt.isEmpty()) {
                log.warn("[Retry] 找不到对应 Email 记录，跳过并保留状态: messageId={}", messageId);
                continue;
            }
            Email email = emailOpt.get();

            try {
                // 标记为“发送中”，并累加重试次数（避免并发节点重复投递）
                ms.setStatus(MessageStatusType.SENT); // 或自定义 RETRYING 枚举
                ms.setRetryCount((ms.getRetryCount() == null ? 0 : ms.getRetryCount()) + 1);
                ms.setUpdateTime(now);
                messageStatusRepo.save(ms);

                // 重新构造事件
                MailInboundEvent event = new MailInboundEvent(
                        messageId,
                        email.getSubject(),
                        email.getFromEmail(),
                        toEmail,
                        toDate(email.getSendTime()),
                        email.getBody()
                );

                // —— 按 fromEmail 分区 —— //
                int partition = computePartition(email.getFromEmail(), partitionCount);
                String routingKey = replyRoutingPrefix + partition;

                // 发到交换机 + routingKey（与生产端一致）
                rabbitTemplate.convertAndSend(replyExchange, routingKey, event);
                log.info("[Retry] 已重投 messageId={} -> exchange={}, routingKey={}",
                        messageId, replyExchange, routingKey);

                // 重投成功后，回到 PENDING，等待消费者处理
                ms.setStatus(MessageStatusType.PENDING);
                ms.setUpdateTime(LocalDateTime.now());
                messageStatusRepo.save(ms);

            } catch (Exception ex) {
                // 重投失败：回到 FAILED（retryCount 已 +1）
                log.error("[Retry] 重投失败 messageId={}, err={}", messageId, ex.getMessage(), ex);
                ms.setStatus(MessageStatusType.FAILED);
                ms.setUpdateTime(LocalDateTime.now());
                messageStatusRepo.save(ms);
            }
        }
    }

    private int computePartition(String fromEmail, int n) {
        if (n <= 0) return 0;
        if (fromEmail == null) return 0;
        return Math.floorMod(fromEmail.hashCode(), n);
    }

    private Date toDate(LocalDateTime ldt) {
        if (ldt == null) return null;
        return Date.from(ldt.atZone(ZoneId.systemDefault()).toInstant());
    }

    /* ========== 清理配置 ========== */

    @Value("${mail.status.cleanup.enabled:true}")
    private boolean cleanupEnabled;

    // 仅删除 updateTime 早于该保留小时数的 ACKED/CONFIRMED 记录
    @Value("${mail.status.cleanup.acked-retention-hours:24}")
    private long ackedRetentionHours;

    // 默认每 10 分钟清理一次
    @Scheduled(cron = "${mail.status.cleanup.cron:0 0/10 * * * ?}")
    @Transactional
    public void purgeAckedMessages() {
        if (!cleanupEnabled) return;

        LocalDateTime cutoff = LocalDateTime.now().minusHours(Math.max(0, ackedRetentionHours));
        long deleted = messageStatusRepo.deleteByStatusAndUpdateTimeBefore(MessageStatusType.CONFIRMED, cutoff);
        if (deleted > 0) {
            log.info("[Cleanup] 已删除 ACKED 记录 {} 条（早于 {}）", deleted, cutoff);
        }
    }
}
